package com.example.flinkcourse.lesson4.source;

import com.example.flinkcourse.lesson4.model.Event;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * 数据源工厂类
 * 实现多种数据源的使用方式
 */
public class SourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(SourceFactory.class);

    /**
     * 创建自定义数据源
     * @return 自定义数据源
     */
    public static SourceFunction<Event> createCustomSource() {
        LOG.info("Creating custom source...");
        return new CustomEventSource();
    }

    /**
     * 创建 Kafka 数据源
     * @param bootstrapServers Kafka 服务器地址
     * @param topic 主题
     * @param groupId 消费者组 ID
     * @return Kafka 数据源
     */
    public static Source<String> createKafkaSource(String bootstrapServers, String topic, String groupId) {
        LOG.info("Creating Kafka source for topic: {}", topic);
        
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("group.id", groupId);
        
        return KafkaSource.<String>builder()
            .setBootstrapServers(bootstrapServers)
            .setTopics(topic)
            .setGroupId(groupId)
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
    }

    /**
     * 创建文件数据源
     * @param inputPath 输入文件路径
     * @param watchInterval 监控间隔（毫秒）
     * @return 文件数据源
     */
    public static Source<String> createFileSource(String inputPath, long watchInterval) {
        LOG.info("Creating file source for path: {}", inputPath);
        return FileSourceFactory.createFileSource(inputPath, watchInterval);
    }

    /**
     * 创建 JDBC 数据源
     * @param url 数据库连接 URL
     * @param query SQL 查询语句
     * @return JDBC 数据源
     */
    public static Source<org.apache.flink.types.Row> createJdbcSource(String url, String query) {
        LOG.info("Creating JDBC source for URL: {}", url);
        return JdbcSourceFactory.createJdbcSource(url, query);
    }

    /**
     * 创建 Socket 数据源
     * @param hostname 主机名
     * @param port 端口
     * @return Socket 数据源
     */
    public static SourceFunction<String> createSocketSource(String hostname, int port) {
        LOG.info("Creating socket source for {}:{}", hostname, port);
        return new org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction(
            hostname,
            port,
            "\n",
            -1
        );
    }

    /**
     * 创建集合数据源
     * @param elements 元素集合
     * @return 集合数据源
     */
    public static SourceFunction<Event> createCollectionSource(Iterable<Event> elements) {
        LOG.info("Creating collection source");
        return new org.apache.flink.streaming.api.functions.source.FromElementsFunction<>(
            elements
        );
    }

    /**
     * 创建并行数据源
     * @param numTasks 并行任务数
     * @return 并行数据源
     */
    public static SourceFunction<Event> createParallelSource(int numTasks) {
        LOG.info("Creating parallel source with {} tasks", numTasks);
        return new ParallelEventSource(numTasks);
    }
}

/**
 * 自定义事件数据源
 */
class CustomEventSource implements SourceFunction<Event> {
    private static final Logger LOG = LoggerFactory.getLogger(CustomEventSource.class);
    private volatile boolean running = true;
    private long count = 0;

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        while (running) {
            Event event = Event.builder()
                .id("event-" + count)
                .type("custom")
                .timestamp(System.currentTimeMillis())
                .amount(Math.random() * 100)
                .build();
            
            ctx.collect(event);
            count++;
            
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

/**
 * 并行事件数据源
 */
class ParallelEventSource implements SourceFunction<Event> {
    private static final Logger LOG = LoggerFactory.getLogger(ParallelEventSource.class);
    private volatile boolean running = true;
    private final int numTasks;
    private long count = 0;

    public ParallelEventSource(int numTasks) {
        this.numTasks = numTasks;
    }

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        while (running) {
            for (int i = 0; i < numTasks; i++) {
                Event event = Event.builder()
                    .id("event-" + count + "-" + i)
                    .type("parallel")
                    .timestamp(System.currentTimeMillis())
                    .amount(Math.random() * 100)
                    .build();
                
                ctx.collect(event);
            }
            count++;
            
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
} 